package com.example.demo.controller;

import com.example.demo.dto.KafkaSearchDto;
import com.example.demo.kafka.KafkaManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

import java.time.Duration;
import java.util.*;

@Slf4j
@RestController
@RequestMapping("/kafka")
public class KafkaController {
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private KafkaManager kafkaManager;

    @PostMapping("/searchMsg")
    public Object searchMsg(@RequestBody KafkaSearchDto dto){
        Properties props = initProperties(dto);

        List<String> list = new ArrayList<>();
        try(KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)){
            Integer offset = dto.getOffset();
            TopicPartition topicPartition = new TopicPartition(dto.getTopic(), 0);
            consumer.assign(Collections.singletonList(topicPartition));

            if(offset != null){
                consumer.seek(topicPartition, dto.getOffset());
            }else{
                consumer.seekToBeginning(Collections.singletonList(topicPartition));
            }
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(dto.getTimeout()));
            for (ConsumerRecord<String, String> record : records) {
                log.info("offset = {}, key = {}, value = {}",
                        record.offset(), record.key(), record.value());
                list.add(record.value());
            }
        }

        return list;
    }

    private static Properties initProperties(KafkaSearchDto dto) {
        Map<String, String> kafkaConfig = dto.getKafkaConfig();
        Properties props = new Properties();
//        props.put("group.id", "test-group");
//        props.put("bootstrap.servers", "localhost:9092");
//        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.putAll(kafkaConfig);
        return props;
    }

    @PostMapping("/put")
    public Object put(@RequestBody KafkaSearchDto dto){
        Properties props = initProperties(dto);
        Producer<String, String> producer = new KafkaProducer<>(props);

        ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key1", dto.getMsg());

        //同步发送
//        RecordMetadata metadata = producer.send(record).get();  // 阻塞等待结果‌:ml-citation{ref="1,7" data="citationList"}

// 异步发送（带回调）
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                log.error("消息发送失败",exception);
            } else {
                log.info("消息发送到分区:{}" ,metadata.partition());
            }
        });
        return 0;

    }

    @PostMapping("/addTopic")
    public Object addTopic(@RequestParam("topic") String topic){
        kafkaManager.addTopic(topic);
        return 0;
    }
    @PostMapping("/removeTopic")
    public Object removeTopic(@RequestParam("topic")String topic){
        kafkaManager.removeTopic(topic);
        return 0;
    }



}
